package cn.lbd.kill.server.service.impl;

import cn.lbd.kill.model.entity.ItemKill;
import cn.lbd.kill.model.entity.ItemKillSuccess;
import cn.lbd.kill.model.mapper.ItemKillMapper;
import cn.lbd.kill.model.mapper.ItemKillSuccessMapper;
import cn.lbd.kill.server.enums.SysConstant;
import cn.lbd.kill.server.service.KillService;
import cn.lbd.kill.server.service.RabbitSenderService;
import cn.lbd.kill.server.utils.RandomUtil;
import cn.lbd.kill.server.utils.SnowFlake;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.framework.recipes.locks.InterProcessReadWriteLock;
import org.joda.time.DateTime;
import org.redisson.Redisson;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.concurrent.TimeUnit;


@Service
@Transactional
public class KillServiceImpl implements KillService {
    private static final Logger log = LoggerFactory.getLogger(KillService.class);
    private static SnowFlake snowFlake = new SnowFlake(2, 3);
    @Autowired(required = false)
    private ItemKillSuccessMapper itemKillSuccessMapper;

    @Autowired(required = false)
    private ItemKillMapper itemKillMapper;

    @Autowired
    private RabbitSenderService rabbitSenderService;

    @Autowired
    private StringRedisTemplate redisTemplate;

    @Autowired
    private Redisson redisson;

    @Autowired
    private CuratorFramework curatorFramework;

    /**
     * 商品秒杀核心业务逻辑的处理
     *
     * @param killId
     * @param userId
     * @return
     * @throws Exception
     */
    @Override
    public Boolean killItem(Integer killId, Integer userId) throws Exception {
        Boolean result = false;
        //判断当前用户是否抢购过当前商品,如果被抢购了，就抢购失败
        if (itemKillSuccessMapper.selectCountByKill(killId, userId) <= 0) {
            //更减库存”时需要判断当前“可被更减的数量”是否仍然还大于0
            ItemKill itemKill = itemKillMapper.getKillItemsInfo(killId);
            if (itemKill != null && 1 == itemKill.getCanKill() && itemKill.getTotal() > 0) {
                //为了阻止并发修改异常，需要在DML语句执行的时候开启数据库的行锁
                int res = itemKillMapper.updateTotalCount(itemKill.getId());
                //扣减成功，就生成秒杀成功的订单，并通过邮件去通知用户
                if (res > 0) {
                    commonRecordKillSuccessInfo(itemKill, userId);
                    result = true;
                }
            }
        } else {
            throw new Exception("您已经抢购过该商品");
        }
        return result;
    }

    /**
     * 商品秒杀核心业务逻辑的优化 ——Redis分布式锁
     *
     * @param killId
     * @param userId
     * @return
     * @throws Exception
     */
    @Override
    public Boolean killItemV2(Integer killId, Integer userId) throws Exception {
        Boolean result = false;
        //将查改增操作包装成原子操作，使用分布式锁
        final String lockKey = new StringBuffer().append(killId).append(userId).append("-RedisLock").toString();
        //用于避免锁失效问题，加入的线程标识
        final String value = RandomUtil.generateOrderCode();
        Boolean flag = redisTemplate.opsForValue().setIfAbsent(lockKey, value);
        //高版本中的setIfAbsent可以同时指定key,value,expire，这里版本低才写开
        redisTemplate.expire(lockKey, 30, TimeUnit.SECONDS);
        try {
            //判断当前用户是否抢购过当前商品,如果被抢购了，就抢购失败
            if (itemKillSuccessMapper.selectCountByKill(killId, userId) <= 0) {
                // redis部署节点宕机了，会出现锁死的情况，需要给这种情况一个退路，于是设置30s后释放锁
                if (flag) {
                    ItemKill itemKill = itemKillMapper.getKillItemsInfo(killId);
                    if (itemKill != null && 1 == itemKill.getCanKill() && itemKill.getTotal() > 0) {
                        //为了阻止并发修改异常，需要在DML语句执行的时候开启数据库的行锁
                        int res = itemKillMapper.updateTotalCount(itemKill.getId());
                        //扣减成功，就生成秒杀成功的订单，并通过邮件去通知用户
                        if (res > 0) {
                            commonRecordKillSuccessInfo(itemKill, userId);
                            result = true;
                        }
                    }
                }
            } else {
                throw new Exception("Redis——您已经抢购过该商品");
            }
        } catch (Exception e) {
            throw new Exception("还没到抢购日期、已过了抢购时间或已被抢购完毕！");
        } finally {
            if (value.equals(redisTemplate.opsForValue().get(lockKey).toString())) {
                redisTemplate.delete(lockKey);
            }
        }
        return result;
    }

    /**
     * 商品秒杀核心业务逻辑的处理,Redisson的分布式锁：获得锁时开一条分线程通过定时器去轮询Redis是否还持有锁，线程标识和和当前的随机数是否一致，是的话给setnx的key延时为初始expire
     *
     * @param killId
     * @param userId
     * @return
     * @throws Exception
     */
    @Override
    public Boolean killItemV3(Integer killId, Integer userId) throws Exception {
        Boolean result = false;
        final String lockKey = new StringBuilder().append(killId).append(userId).append("_RedissonLock").toString();
        RLock lock = redisson.getLock(lockKey);
        try {
            // 人性化时间，可重入锁，30s尝试，获取锁后10s释放
            //boolean cacheRes = lock.tryLock(30, 10, TimeUnit.SECONDS);
            lock.lock(10,TimeUnit.SECONDS);
            //if (cacheRes) {
                //判断当前用户是否抢购过当前商品,如果被抢购了，就抢购失败
                if (itemKillSuccessMapper.selectCountByKill(killId, userId) <= 0) {
                    //更减库存”时需要判断当前“可被更减的数量”是否仍然还大于0
                    ItemKill itemKill = itemKillMapper.getKillItemsInfo(killId);
                    if (itemKill != null && 1 == itemKill.getCanKill() && itemKill.getTotal() > 0) {
                        //为了阻止并发修改异常，需要在DML语句执行的时候开启数据库的行锁
                        int res = itemKillMapper.updateTotalCount(itemKill.getId());
                        //扣减成功，就生成秒杀成功的订单，并通过邮件去通知用户
                        if (res > 0) {
                            commonRecordKillSuccessInfo(itemKill, userId);
                            result = true;
                        }
                    }
                } else {
                    throw new Exception("您已经抢购过该商品");
                }
            //}
        } catch (Exception e) {
            throw new Exception("还没到抢购日期、已过了抢购时间或已被抢购完毕！");
        } finally {
            lock.unlock();
            //强制释放
            //lock.forceUnlock();
        }
        return result;
    }

    /**
     * 商品秒杀核心业务逻辑的优化，Zookeeper利用临时有序结点实现分布式锁
     *
     * @param killId
     * @param userId
     * @return
     * @throws Exception
     */
    private final String zkPrefix = "/kill";
    @Override
    public Boolean killItemV4(Integer killId, Integer userId) throws Exception {
        Boolean result = false;
        // 进程之间的互斥操作锁
        InterProcessMutex interProcessMutex = new InterProcessMutex(curatorFramework,zkPrefix + killId + userId);
        try {
            //获取锁
            if(interProcessMutex.acquire(10L,TimeUnit.SECONDS)){
                //判断当前用户是否抢购过当前商品,如果被抢购了，就抢购失败
                if (itemKillSuccessMapper.selectCountByKill(killId, userId) <= 0) {
                    //更减库存”时需要判断当前“可被更减的数量”是否仍然还大于0
                    ItemKill itemKill = itemKillMapper.getKillItemsInfo(killId);
                    if (itemKill != null && 1 == itemKill.getCanKill() && itemKill.getTotal() > 0) {
                        //为了阻止并发修改异常，需要在DML语句执行的时候开启数据库的行锁
                        int res = itemKillMapper.updateTotalCount(itemKill.getId());
                        //扣减成功，就生成秒杀成功的订单，并通过邮件去通知用户
                        if (res > 0) {
                            commonRecordKillSuccessInfo(itemKill, userId);
                            result = true;
                        }
                    }
                } else {
                    throw new Exception("您已经抢购过该商品");
                }
            }
        }catch (Exception e) {
            throw new Exception("还没到抢购日期、已过了抢购时间或已被抢购完毕！");
        }finally {
            if (interProcessMutex != null) {
                interProcessMutex.release();
            }
        }

        return result;
    }


    /**
     * 通用的方法-记录用户秒杀成功后生成的订单-并进行异步邮件消息的通知
     */
    private void commonRecordKillSuccessInfo(ItemKill kill, Integer userId) {
        ItemKillSuccess itemKillSuccess = new ItemKillSuccess();
        //利用 时间戳加N位随机数流水号 保证分布式id的唯一性
        //itemKillSuccess.setCode(RandomUtil.generateOrderCode());
        //利用 雪花算法 保证分布式id的唯一性
        String orderNo = String.valueOf(snowFlake.nextId());
        itemKillSuccess.setCode(orderNo);
        itemKillSuccess.setItemId(kill.getItemId());
        itemKillSuccess.setKillId(kill.getId());
        //秒杀成功未支付
        itemKillSuccess.setStatus(SysConstant.OrderStatus.SuccessNotPayed.getCode().byteValue());
        itemKillSuccess.setUserId(userId.toString());
        itemKillSuccess.setCreateTime(DateTime.now().toDate());
        // 仿照单例模式的双重检验锁写法
        // 查完更新还是会出差错(写入操作不能保证每次都为查询操作争取足够多的时间),还是要将这一系列操作变成原子操作才行
        if (itemKillSuccessMapper.selectCountByKill(kill.getId(), userId) <= 0) {
            //生成秒杀订单
            log.info("生成秒杀订单");
            int res = itemKillSuccessMapper.insert(itemKillSuccess);
            if (res > 0) {
                //进行异步邮件的消息通知
                rabbitSenderService.sendKillSuccessEmailMsg(orderNo);
                //将消息入死信队列，用于 “失效” 超过指定的TTL时间时仍然未支付的订单
                rabbitSenderService.sendExpireKillSuccessMsg(orderNo);
            }
        }
    }
}
